package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.KwSplit;
import com.atguigu.gmall.realtime.util.FlinkSinUtil;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/2/17 08:30
 */
public class Dws_01_DwsTrafficKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new Dws_01_DwsTrafficKeywordPageViewWindow().init(
            4001,
            2,
            "Dws_01_DwsTrafficKeywordPageViewWindow"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取页面日志数据
        tEnv.executeSql("create table page(" +
                            " page map<string, string>, " +
                            " ts bigint, " +
                            " et as to_timestamp_ltz(ts, 3), " +
                            " watermark for et as et - interval '3' second " +
                            ")" + SQLUtil.getDDLKafkaSource(Constant.TOPIC_DWD_TRAFFIC_PAGE, "Dws_01_DwsTrafficKeywordPageViewWindow"));
        
        // 2. 过滤出搜索记录
        Table kwTable = tEnv.sqlQuery("select " +
                                          " page['item'] kw, " +
                                          " et " +
                                          "from page " +
                                          "where (page['last_page_id']='home' or page['last_page_id']='search') " +
                                          "and page['item'] is not null " +
                                          "and page['item_type']='keyword'");
        tEnv.createTemporaryView("kw_table", kwTable);
        
        // 3. 对关键词进行分词
        // 自定义TableFunction
        // 3.1 先注册自定义函数
        tEnv.createTemporaryFunction("kw_split", KwSplit.class);
        // 3.2 在 sql 中使用 TableFunction
        Table keywordTable = tEnv.sqlQuery("select " +
                                               " keyword, " +
                                               " et " +
                                               "from kw_table " +
                                               "join lateral table(kw_split(kw))  on true");
        tEnv.createTemporaryView("keyword_table", keywordTable);
        
        // 4. 开窗聚会:  tvf
        Table result = tEnv.sqlQuery("select " +
                                         "date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                                         "date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                                         "keyword, " +
                                         "count(*) keyword_count, " +
                                         "unix_timestamp() * 1000 ts " +  // count(*)  sum(1) count(1)   count(id)
                                         "from table(tumble(table keyword_table, descriptor(et), interval '5' second ))" +
                                         "group by window_start, window_end, keyword ");
        
        // 5. 写出到 clickhouse 中
        // 自定义 sink , 转成流之后, 再自定义
        SingleOutputStreamOperator<KeywordBean> resultStream = tEnv
            .toRetractStream(result, KeywordBean.class)
            .filter(t -> t.f0)
            .map(t -> t.f1);
    
        resultStream.addSink(FlinkSinUtil.getClickHouseSink("dws_traffic_keyword_page_view_window", KeywordBean.class));
    
    
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        
        
    }
}
/*
group by key
流量域 搜索关键词粒度 页面浏览各窗口汇总表
窗口      关键词         次数
0-5       手机           10
0-5       电脑           20
...
5-10      电脑           30
...

流技术

sql 技术

1. 通过 ddl 从 kafka 的 dwd 层topic的读取数据
    页面日志

2. 过滤出搜索记录, 找到搜索的关键词
    // TODO
    1. 首页输入关键词, 回车, 进入搜索的结果页面
        首页->搜索结果页面
       last_page_id=home && item_type=keyword && item is not null
        
    2. 从专门的搜索页面, 输入关键词, 回车进入结果要么
       last_page_id=search && item_type=keyword && item is not null
        
      合并:
        (last_page_id=home || last_page_id=search) && item_type=keyword && item is not null
        
        
     "手机"
     "华为白色128g手机"
     "华为黑色256g手机5G"
     
     按照中文的习惯,对关键词进行分词
     
     "手机"
     "华为
     白色
     128g
     手机"
     "华为
     黑色
     256g
     手机
     5G"
     
     如何分词:
        ik 分词器
      
      列转行
        自定义函数
            scalar  一进一出
            aggregate 多进一出
            table     一进多出
            tableAggregate 多进多出
    
     
     
     
    
    
3. 开窗聚合
    grouped window: 过时的, 不建议使用
    
    tvf
        滚动
        滑动
        累积
        
4. 数据写出到 clickhouse 中
    自定义 sink
        



------------
dws 层:
    汇总: 小窗口汇总
    
    0-5 ...
    5-10 ...
    
    写入到 clickhouse 中.
    
    
有些公司, 确实没有 dws 层, 直接把 dwd 的明细数据, 写入到 olap 数据(clickhouse, doris)


*/